package com.atguigu.flink.exec2;

import com.atguigu.flink.function.MyUtil;
import com.atguigu.flink.pojo.OrderEvent;
import com.atguigu.flink.pojo.UserBehavior;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.EventTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.List;

/**
 * Created by Smexy on 2022/12/17
 *
 * 需求: 找到超时(30min)和对账失败(只有支付信息没有订单信息)的订单。
 *
 *  使用下单和支付进行比对。
 *
 *          ①下单 ----支付 ， 间隔(gap)30min以内 ，正常
 *          ②下单 ----支付 ， 间隔30min以外 ，不正常
 *          ③下单                          正常
 *          ④支付                          不正常
 *
 *
 *     按照orderId分组，把同一笔单的支付和下单分到一起。
 *
 *          30min：  时间窗口。
 *                      session窗口
 *
 *     -----------
 *      窗口触发运算：
 *              有以下可能:
 *                      ①有两条数据。
 *                             下单 ----支付 ，间隔(gap)30min以内 ，正常
 *
 *                      ②只有1条数据
 *                              是下单，需要把自己写入状态。
 *                                  仅有下单
 *                                  支付超时
 *
 *                              是支付。从状态中取下单，取到，就是超时，否则就是缺失下单
 *                                  支付超时
 *                                  支付缺失下单
 *
 *
 *
 */
public class Demo6_Order
{
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        WatermarkStrategy<OrderEvent> watermarkStrategy = WatermarkStrategy.<OrderEvent>forMonotonousTimestamps()
            .withTimestampAssigner( (e, r) -> e.getEventTime() * 1000);

        env.setParallelism(1);

        env.readTextFile("data/OrderLog.csv")
            .map(new MapFunction<String, OrderEvent>()
            {
                @Override
                public OrderEvent map(String value) throws Exception {
                    String[] words = value.split(",");
                    return new OrderEvent(
                        Long.valueOf(words[0]),
                        words[1],
                        words[2],
                        Long.valueOf(words[3])
                    );
                }
            })
           .assignTimestampsAndWatermarks(watermarkStrategy)
           .keyBy(OrderEvent::getOrderId)
           .window(EventTimeSessionWindows.withGap(Time.minutes(30)))
           .process(new ProcessWindowFunction<OrderEvent, String, Long, TimeWindow>()
           {

               private ValueState<String> orderId;

               @Override
               public void open(Configuration parameters) throws Exception {
                   orderId = getRuntimeContext().getState(new ValueStateDescriptor<>("orderId", String.class));
               }

               // elements 可能有2个，可能只有1个
               @Override
               public void process(Long aLong, Context context, Iterable<OrderEvent> elements, Collector<String> out) throws Exception {

                   List<OrderEvent> orderEvents = MyUtil.parseIterable(elements);

                   if (orderEvents.size() == 1){

                       OrderEvent orderEvent = orderEvents.get(0);

                       //可能是下单
                       if ("create".equals(orderEvent.getEventType())){

                           orderId.update(orderEvent.getOrderId().toString());

                       }else{

                           //是支付
                           String orderIdStr = orderId.value();

                           if (orderIdStr == null){
                               out.collect(orderEvent.getOrderId() +"  缺少create信息");
                           }else {
                               out.collect(orderIdStr + "支付超时...");
                           }

                       }

                   }
               }
           })
           .print();

        env.execute();

    }
}
